use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::process::Output;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};
use crossbeam::channel;
use futures::task;
use futures::task::ArcWake;
use tokio::macros::support::poll_fn;
use tokio::sync::Notify;
use crate::gen_title;

pub async fn ch7_2_1() {
    gen_title("ch7_2_1", 18);

    struct Delay {
        when: Instant,
    }

    impl Future for Delay {
        type Output = &'static str;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if Instant::now() >= self.when {
                println!("Hello world");
                Poll::Ready("done")
            } else {
                cx.waker().wake_by_ref();
                Poll::Pending
            }
        }
    }

    let when = Instant::now() + Duration::from_millis(10);
    let future = Delay { when };

    let out = future.await;

    assert_eq!(out, "done");
    println!("{}", out);
}

pub async fn ch7_2_2() {
    gen_title("ch7_2_2", 18);

    struct MiniTokio {
        scheduled: channel::Receiver<Arc<Task>>,
        sender: channel::Sender<Arc<Task>>,
    }

    struct Task {
        future: Mutex<Pin<Box<dyn Future<Output=()> + Send>>>,
        executor: channel::Sender<Arc<Task>>,
    }

    impl Task {
        fn schedule(self: &Arc<Self>) {
            self.executor.send(self.clone());
        }

        fn poll(self: Arc<Self>) {
            let waker = task::waker(self.clone());
            let mut cx = Context::from_waker(&waker);

            let mut future = self.future.try_lock().unwrap();

            let _ = future.as_mut().poll(&mut cx);
        }

        fn spawn<F>(future: F, sender: &channel::Sender<Arc<Task>>) where F: Future<Output=()> + Send + 'static {
            let task = Arc::new(Task {
                future: Mutex::new(Box::pin(future)),
                executor: sender.clone(),
            });

            let _ = sender.send(task);
        }
    }

    impl ArcWake for Task {
        fn wake_by_ref(arc_self: &Arc<Self>) {
            arc_self.schedule();
        }
    }

    impl MiniTokio {
        fn new() -> MiniTokio {
            let (sender, scheduled) = channel::unbounded();
            MiniTokio {
                scheduled,
                sender,
            }
        }

        fn spawn<F>(&mut self, future: F) where F: Future<Output=()> + Send + 'static {
            Task::spawn(future, &self.sender);
        }

        fn run(&mut self) {
            while let Ok(task) = self.scheduled.recv() {
                task.poll();
            }
        }
    }

    struct Delay {
        when: Instant,
    }

    impl Future for Delay {
        type Output = &'static str;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if Instant::now() >= self.when {
                println!("Hello world");
                Poll::Ready("done")
            } else {
                let waker = cx.waker().clone();
                let when = self.when;

                thread::spawn(move || {
                    let now = Instant::now();

                    if now < when {
                        thread::sleep(when - now);
                    }

                    waker.wake();
                });

                Poll::Pending
            }
        }
    }

    let when = Instant::now() + Duration::from_millis(10);
    let mut delay = Some(Delay { when });

    poll_fn(move |cx| {
        let mut delay = delay.take().unwrap();
        let res = Pin::new(&mut delay).poll(cx);
        assert!(res.is_pending());
        tokio::spawn(async move {
            delay.await;
        });

        Poll::Ready(())
    }).await
}

pub async fn ch7_2_3() {
    gen_title("ch7_2_3", 18);

    struct Delay {
        when: Instant,
        waker: Option<Arc<Mutex<Waker>>>,
    }

    impl Future for Delay {
        type Output = ();

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if let Some(waker) = &self.waker {
                let mut waker = waker.lock().unwrap();
                if !waker.will_wake(cx.waker()) {
                    *waker = cx.waker().clone();
                }
            } else {
                let when = self.when;
                let waker = Arc::new(Mutex::new(cx.waker().clone()));
                self.waker = Some(waker.clone());

                thread::spawn(move || {
                    let now = Instant::now();

                    if now < when {
                        thread::sleep(when - now);
                    }

                    let waker = waker.lock().unwrap();
                    waker.wake_by_ref();
                });
            }

            if Instant::now() >= self.when {
                println!("Hello world");
                Poll::Ready(())
            } else {
                Poll::Pending
            }
        }
    }

    let when = Instant::now() + Duration::from_millis(10);
    let mut delay = Some(Delay { when, waker: None });

    poll_fn(move |cx| {
        let mut delay = delay.take().unwrap();
        let res = Pin::new(&mut delay).poll(cx);
        assert!(res.is_pending());
        tokio::spawn(async move {
            delay.await;
        });

        Poll::Ready(())
    }).await
}

pub async fn ch7_2_4() {
    gen_title("ch7_2_4", 18);

    let when = Instant::now() + Duration::from_millis(10);

    let notify = Arc::new(Notify::new());
    let notify2 = notify.clone();

    thread::spawn(move || {
        let now = Instant::now();

        if now < when {
            thread::sleep(when - now);
        }

        println!("hello world");

        notify2.notify_one();
    });

    notify.notified().await;
}